//#define _GNU_SOURCE // This is necessary for the code to compile. Add it to the very top of your program.

#include <sstream>
#include <string>
#include <list>
#include <iostream>
#include <fstream>
#include <cstdlib>
#include <pthread.h>
#include <vector>
#include <errno.h>
#include <ctime>
#include <unistd.h>
#include <cmath>
#include <sys/time.h>
#include <iomanip>
#include <sched.h>

//extern vector<int> testvector;

using namespace std;

//Main Thread
//Reads in Data to string stream
//Saves data into list
//Iterates through list and breaks into chunks
//putting descriptor and index of beginning and end
//into an array and having a linked list of arrays

double Timestamp(void); // returns current time
void c2s(string &old); //turns commas into spaces for delimmeter of string stream
void datafile2list(); //puts data file into list of integers
void *subthread(void *stuff);
void *mergethread(void *stuff);
list<int> list1; //create list of integers
int counter = 1; //size of array of data file integers
list<int> chunklist; //list that holds the information for the splitting of chunks
vector<int> fullvector;
int chunksize = 0;
int z=0;
int v=0;
int P; //P threads  -- comes from arguments passed when function is run
int C; //C cores    -- comes from arguments passed when function is run
int limit_cores(pthread_t *thread, int num_cores); //limit cores function
vector<pthread_t> threadvector;

pthread_mutex_t count_mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t barrier_cond = PTHREAD_COND_INITIALIZER;

double start_time, finish_time;

int main(int argc, char* argv[])
{
    start_time = Timestamp();
    pthread_t tid;

    if (argc != 3)
    {
		cout << "usage: a.out <No. Threads> <No. Cores>\n";
		return -1;
    }

    P = atoi(argv[1]);
    C = atoi(argv[2]);
    
    threadvector.resize(P);
    
    //pthread_t thread = pthread_self();
    //limit_cores(&thread, C);

    datafile2list();
    //Starts the chunk making
    int count = 1;
    int arraychunk[3];
    int M;

    arraychunk[0] = count;
    arraychunk[1] = 0;
   
    for(int i = 2; i <= counter; i++)
    {
        if(i % 1024 == 1)
        {
            arraychunk[1] = i-1;
            count++;
            arraychunk[0] = count;
        }	
        else if(i % 1024 == 0)
        {
            arraychunk[2] = i-1;
            chunklist.push_back(arraychunk[0]);
            chunklist.push_back(arraychunk[1]);
            chunklist.push_back(arraychunk[2]);
        }
    }
    chunksize = chunklist.size();
    M = chunksize / 3;
    list<int>::iterator j;

    cout << "Reading " << M << " data chunks..." << endl;
    cout << Timestamp() << "Create " << P << " Threads..." << endl;
    cout << "Binding threads to " << C << " CPU cores..." << endl;
    
    while(!chunklist.empty())
    {
        for(int g = 0; g < P; g++)
        {
            pthread_create(&tid, NULL, subthread, NULL);
            limit_cores(&tid, C);
            pthread_join(tid, NULL);
	    cout << Timestamp() << "No more work to do. Thread " << g << "exiting..." << endl;
        }
    }
   
    cout << "Merge start " << Timestamp() << endl;
    cout << "Creating " << M/2 << " " << Timestamp() << endl;
//    for(int u = 1; u <= log2(chunksize / 3); u++)
//    {
	while(z <= M/2)
    	{
    	    pthread_create(&tid, NULL, mergethread, NULL);
            limit_cores(&tid, C);
//    	    pthread_cond_wait(&barrier_cond, &count_mutex);
    	    pthread_join(tid, NULL);
    	}
//        M = M / 2;
//	z = 0;
//    }
    cout << "All threads finished " << Timestamp() << endl;
    
    pthread_exit(0);
    cout << "Main thread exit " << Timestamp() << endl;
    system("rm -f file*");
    finish_time = Timestamp();
    cout << "The program finished in " << finish_time - start_time << " with " << C << " number of cores." << endl;
    return 0;
}

void datafile2list()
{
    stringstream sstream;
    ifstream f("data.in");
    sstream << f.rdbuf();
    string file = sstream.str();
    c2s(file);

    stringstream a;
    a << file;

    while(!(a.eof()))
    {
            int currentint;
            a >> currentint;
            fullvector.push_back(currentint);
    }
    f.close();	
}

void c2s(string &old) //turns commas into spaces for delimmeter of string stream
{
    counter = 1;
    int prev = 0;
    int prev2 = 0;
    while(1)
    {
        prev = old.find(',', 0); //need to change commas...
        if(prev != string::npos)
        {
            old.replace(prev, 1," "); //to spaces
            counter++;
        }
        else
            break;
    }
    while(1)
    {
        prev2 = old.find('\n', 0);
        if(prev2 != string::npos)
            old.replace(prev2, 1, "\0");
        else
            break;
    }
}

void *subthread(void *stuff)
{
    if(chunklist.empty()) //if it's empty, return and don't do anything
        return NULL;
        
    list<int> extractedData; //list where chunk actual data is stored temporarily
    int chunknum;
    int firstindex;
    int lastindex;
    
    pthread_mutex_lock(&count_mutex);
    cout << "Thread starts work on chunk " << Timestamp() << endl;
    list<int>::iterator j;
    
    j = chunklist.begin();
    chunknum = *j;
    chunklist.pop_front();
    *j = chunklist.front();
    firstindex = *j;
    chunklist.pop_front();
    *j = chunklist.front();
    lastindex = *j;
    chunklist.pop_front();
    *j = chunklist.front();
    for( int i = firstindex; i < lastindex; i++)
        extractedData.push_back(fullvector[i]);

    extractedData.sort();
    
    list<int>::iterator k;
    
    ofstream outfile;

    stringstream filename;

    filename << "file" << chunknum;

    outfile.open(filename.str().c_str());
        
    for(k = extractedData.begin(); k != extractedData.end(); k++)
        outfile << *k << " ";
        
    outfile.close();

    cout << "Thread finished work on chunk " << Timestamp() << endl;
    pthread_mutex_unlock(&count_mutex);
}

void *mergethread(void *stuff)
{
    v++;
    cout << "Merge thread start working on file " << Timestamp() << endl;
    list<int> mergelist;
    
    mergelist.clear();
    
    stringstream sfile1;
    stringstream sfile2;
    
    stringstream sstream1;
    stringstream sstream2;
    
    stringstream delfile1;
    stringstream delfile2;
        
    z++;
    sfile1 << "file" << z;
    z++;
    sfile2 << "file" << z;
    
    ifstream f1(sfile1.str().c_str());
    ifstream f2(sfile2.str().c_str());
    
    sstream1 << f1.rdbuf();
    sstream2 << f2.rdbuf();
           
    while(!(sstream1.eof()))
    {
            int currentint1;
            sstream1 >> currentint1;
            mergelist.push_back(currentint1);
    }
    
    while(!(sstream2.eof()))
    {
            int currentint2;
            sstream2 >> currentint2;
            mergelist.push_back(currentint2);
    }
            
    f1.close();
    f2.close();
    
    delfile1 << "rm " << sfile1.str();
    delfile2 << "rm " << sfile2.str();
    
    system(delfile1.str().c_str());
    system(delfile2.str().c_str());
    
    mergelist.sort();

    ofstream mergefile;

    stringstream mergefilename;

    mergefilename << "file" << v;

    mergefile.open(mergefilename.str().c_str());

    list<int>::iterator l;
    
    for(l = mergelist.begin(); l != mergelist.end(); l++)
        mergefile << *l << " ";
        
    mergefile.close();
    cout << "Merge thread exiting " << Timestamp() << endl;
}


// Copy this function into your code.
// You use it by passing it a pointer to a pthread_t and the number of cores
// that you want the thread limited to.

int limit_cores(pthread_t *thread, int num_cores)
{
  int s, j;
  cpu_set_t cpuset;

  // Set affinity mask to include CPUs 0 to (num_cores-1)

  CPU_ZERO(&cpuset);
  for (j = 0; j < num_cores; j++)
     CPU_SET(j, &cpuset);

  s = pthread_setaffinity_np(*thread, sizeof(cpu_set_t), &cpuset);
  if(s != 0)
  {
    printf("Error %i in pthread_setaffinity_np()...\n", s);
    return -1;
  }

  // Check the actual affinity mask assigned to the thread
  s = pthread_getaffinity_np(*thread, sizeof(cpu_set_t), &cpuset); 
  if(s != 0)
  {
    printf("Error %i in pthread_getaffinity_np()...\n", s);
    return -1;
  }

  // Check to see how many cores are assigned to the thread.
  int count = 0;
  for (j = 0; j < CPU_SETSIZE; j++)
  if (CPU_ISSET(j, &cpuset))
    count++;

  if(count > num_cores)
  {
    printf("Error, the thread seems to be using more cores than you requested...\n");
    return -2;
  }

  return 0;
}

double Timestamp() {
  struct timeval tp;
  static double last_timestamp = 0.0;
  double timestamp;
  gettimeofday (&tp, 0);
  timestamp = (double) tp.tv_sec * 1e6 + tp.tv_usec;
  return timestamp;
}

